package com.chb.flink.state

import com.chb.flink.source.StationLog
import org.apache.flink.api.common.functions.RichFlatMapFunction
import org.apache.flink.api.common.state.{ValueState, ValueStateDescriptor}
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
import org.apache.flink.util.Collector


/**
 * 案例需求：计算每个手机的呼叫间隔时间，单位是毫秒
 * 第一种方法
 */
object TestKeyState1 {
    def main(args: Array[String]): Unit = {
        val streamEnv = StreamExecutionEnvironment.getExecutionEnvironment
        import org.apache.flink.streaming.api.scala._

        //读取文件数据
        val data = streamEnv.readTextFile(getClass.getResource("/station.log").getPath)
            .map(line => {
                var arr = line.split(",")
                new StationLog(arr(0).trim, arr(1).trim, arr(2).trim, arr(3).trim, arr(4).trim.toLong, arr(5).trim.toLong)
            })

        data.keyBy(_.callOut) // 按照主键分组
                .flatMap(new CallIntervalFunction())
                .print()


        streamEnv.execute()

    }

    /**
     * 记录通话的时间间隔
     */
    class CallIntervalFunction() extends RichFlatMapFunction[StationLog, (String, Long)] {
        var preCallTimeState: ValueState[Long] = _

        override def open(parameters: Configuration): Unit = {
            preCallTimeState = getRuntimeContext.getState(new ValueStateDescriptor[Long]("pre", classOf[Long]))
        }

        // 业务逻辑
        override def flatMap(in: StationLog, collector: Collector[(String, Long)]): Unit = {
            var pre = preCallTimeState.value()
            if (pre == null || pre == 0) { // 第一次呼叫
                preCallTimeState.update(in.callTime)
            } else {
                val interval = in.callTime - pre
                collector.collect((in.callOut, interval))
                preCallTimeState.update(in.callTime)
            }
        }
    }




}
